package com.xrui.hbase.impl;

import com.google.common.base.Preconditions;

import com.xrui.hbase.ColumnName;
import com.xrui.hbase.DataRequest;
import com.xrui.hbase.EntityId;
import com.xrui.hbase.HBaseEntityId;
import com.xrui.hbase.schema.TableLayout;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Map;
import java.util.NavigableSet;

/**
 * Wraps a DataRequest to expose methods that generate meaningful objects in HBase
 * land, like {@link org.apache.hadoop.hbase.client.Put}s and {@link
 * org.apache.hadoop.hbase.client.Get}s.
 */
public final class HBaseDataRequestAdapter {
    private static final Logger LOG = LoggerFactory.getLogger(HBaseDataRequestAdapter.class);

    /**
     * The wrapped DataRequest.
     */
    private final DataRequest mDataRequest;
    /** The translator for generating HBase column names. */
    //private final HBaseColumnNameTranslator mColumnNameTranslator;

    /**
     * Creates a new HBaseDataRequestAdapter for a given data request using a given
     * ColumnNameTranslator.
     *
     * @param dataRequest the data request to adapt for HBase.
     */
    public HBaseDataRequestAdapter(
        final DataRequest dataRequest
    ) {
        mDataRequest = dataRequest;
    }

    /**
     * Adds a fully-qualified column to an HBase Get request, if necessary.
     * <p>
     * <p>
     * If the entire HBase family is already requested, the column does not need to be added.
     * </p>
     *
     * @param get    Adds the column to this Get request.
     * @param column Fully-qualified HBase column to add to the Get request.
     * @return the Get request.
     */
    private static Get addColumn(final Get get, final ColumnName column) {
        // Calls to Get.addColumn() invalidate previous calls to Get.addFamily(),
        // so we only do it if:
        //   1. No data from the family has been added to the request yet,
        // OR
        //   2. Only specific columns from the family have been requested so far.
        // Note: the Get family-map uses null values to indicate requests for an entire HBase family.
        if (!get.familySet().contains(column.getFamilyBytes())
            || (get.getFamilyMap().get(column.getFamilyBytes()) != null)) {
            get.addColumn(column.getFamilyBytes(), column.getQualifierBytes());
        }
        return get;
    }

    /**
     * Constructs and returns the HBase filter that returns only the
     * data in a given column request.
     *
     * @param columnRequest A column request.
     * @return An HBase filter that retrieves only the data for the column request.
     * @throws IOException If there is an error.
     */
    private static Filter toFilter(
        final DataRequest.Column columnRequest
    ) throws IOException {

        final ColumnName columnName = columnRequest.getColumnName();

        final FilterList filter = new FilterList(FilterList.Operator.MUST_PASS_ALL);

        // 1. FamilyFilter
        filter.addFilter(new FamilyFilter(CompareFilter.CompareOp.EQUAL,
            new BinaryComparator(columnName.getFamilyBytes())));

        // 2. QualifierFilter
        if (columnName.isFullyQualified()) {
            filter.addFilter(new QualifierFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryComparator(columnName.getQualifierBytes())));
        }

        // 3. ColumnPaginationFilter when paging or full-qualified max-versions is 1
        if (columnRequest.isPagingEnabled()
            || (columnName.isFullyQualified() && (columnRequest.getMaxVersions() == 1))) {
            // For fully qualified columns where maxVersions = 1, we can use the
            // ColumnPaginationFilter to restrict the number of versions returned to at most 1.
            //
            // Other columns' maxVersions will be filtered client-side in HBaseRowData.
            //
            // Prior to HBase 0.94, we could use this optimization for all fully qualified
            // columns' maxVersions requests, due to different behavior in the
            // ColumnPaginationFilter.
            //
            // Note: we could also use this for a map-type family if max-versions == 1,
            //     by setting limit = Integer.MAX_VALUE.
            final int limit = 1;
            final int offset = 0;
            filter.addFilter(new ColumnPaginationFilter(limit, offset));
        }

        //4. User-specified column filter, if specified:
        if (columnRequest.getFilter() != null) {
            filter.addFilter(
                columnRequest.getFilter().toHBaseFilter(columnName));
        }

        //5. FirstKeyOnlyFilter. If column has paging enabled, we just want to know about the existence of a cell:
        if (columnRequest.isPagingEnabled()) {
            filter.addFilter(new FirstKeyOnlyFilter());

            // TODO(SCHEMA-334) KeyOnlyFilter uses Filter.transform() which applies unconditionally.
            //     There is a chance that Filter.transform() may apply conditionally in the future,
            //     in which case we may re-introduce the KeyOnlyFilter.
            //     An alternative is to provide a custom HBase filter to handle data requests
            //     efficiently.
        }

        return filter;
    }

    /**
     * Constructs an HBase Scan that describes the data requested in the DataRequest.
     *
     * @param tableLayout The layout of the table to read from.  This is required for
     *                    determining the mapping between columns and HBase columns.
     * @return An HBase Scan descriptor.
     * @throws IOException If there is an error.
     */
    public Scan toScan(final TableLayout tableLayout) throws IOException {
        return toScan(tableLayout, new HBaseScanOptions());
    }

    /**
     * Constructs an HBase Scan that describes the data requested in the DataRequest.
     *
     * @param tableLayout The layout of the table to read from.  This is required for
     *                    determining the mapping between columns and HBase columns.
     * @param scanOptions Custom options for this scan.
     * @return An HBase Scan descriptor.
     * @throws IOException If there is an error.
     */
    public Scan toScan(
        final TableLayout tableLayout,
        final HBaseScanOptions scanOptions
    ) throws IOException {
        // Unfortunately in HBase 95+, we can no longer create empty gets.
        // So create a fake one for this table and fill in the fields of a new scan.
        final Get tempGet = toGet(HBaseEntityId.create(new byte[1], tableLayout), tableLayout);
        final Scan scan = new Scan();
        scan.setFilter(tempGet.getFilter());
        scan.setCacheBlocks(tempGet.getCacheBlocks());
        scan.setMaxVersions(tempGet.getMaxVersions());
        scan.setTimeRange(tempGet.getTimeRange().getMin(), tempGet.getTimeRange().getMax());
        scan.setFamilyMap(tempGet.getFamilyMap());
        configureScan(scan, scanOptions);
        return scan;
    }

    /**
     * Like toScan(), but mutates a given Scan object to include everything in the data
     * request instead of returning a new one.
     * <p>
     * <p>Any existing request settings in the Scan object will be preserved.</p>
     *
     * @param scan        The existing scan object to apply the data request to.
     * @param tableLayout The layout of the table the scan will read from.
     * @throws IOException If there is an error.
     */
    public void applyToScan(
        final Scan scan,
        final TableLayout tableLayout
    ) throws IOException {
        final Scan newScan = toScan(tableLayout);

        // It's okay to put columns into the Scan that are already there.
        for (Map.Entry<byte[], NavigableSet<byte[]>> columnRequest
            : newScan.getFamilyMap().entrySet()) {
            byte[] family = columnRequest.getKey();
            if (null == columnRequest.getValue()) {
                // Request all columns in the family.
                scan.addFamily(family);
            } else {
                // Calls to Scan.addColumn() will invalidate any previous calls to Scan.addFamily(),
                // so we only do it if:
                //   1. No data from the family has been added to the request yet, OR
                //   2. Only specific columns from the family have been requested so far.
                if (!scan.getFamilyMap().containsKey(family)
                    || null != scan.getFamilyMap().get(family)) {
                    for (byte[] qualifier : columnRequest.getValue()) {
                        scan.addColumn(family, qualifier);
                    }
                }
            }
        }
    }

    /**
     * Constructs an HBase Get that describes the data requested in the DataRequest for
     * a particular entity/row.
     *
     * @param entityId    The row to build an HBase Get request for.
     * @param tableLayout The layout of the table to read from.  This is required for
     *                    determining the mapping between columns and HBase columns.
     * @return An HBase Get descriptor.
     * @throws IOException If there is an error.
     */
    public Get toGet(
        final EntityId entityId,
        final TableLayout tableLayout
    ) throws IOException {

        // Context to translate user filters into HBase filters:
        //final ColumnFilter.Context filterContext =
        //   new NameTranslatingFilterContext(mColumnNameTranslator);

        // Get request we are building and returning:
        final Get get = new Get(entityId.getHBaseRowKey());

        // Filters for each requested column: OR(<filter-for-column-1>, <filter-for-column2>, ...)
        final FilterList columnFilters = new FilterList(FilterList.Operator.MUST_PASS_ONE);

        // There's a shortcoming in the HBase API that doesn't allow us to specify per-column
        // filters for timestamp ranges and max versions.  We need to generate a request that
        // will include all versions that we need, and add filters for the individual columns.

        // As of HBase 0.94, the ColumnPaginationFilter, which we had been using to permit per-column
        // maxVersions settings, no longer pages over multiple versions for the same column. We can
        // still use it, however, to limit fully-qualified columns with maxVersions = 1 to return only
        // the most recent version in the request's time range. All other columns will use the largest
        // maxVersions seen on any column request.

        // Fortunately, although we may retrieve more versions per column than we need from HBase, we
        // can still honor the user's requested maxVersions when returning the versions in
        // HBaseRowData.

        // Largest of the max-versions from all the requested columns.
        // Columns with paging are excluded (max-versions does not make sense when paging):
        int largestMaxVersions = 1;

        // If every column is paged, we should add a keyonly filter to a single column, so we can have
        // access to entityIds in our HBaseRowData that is constructed.
        boolean completelyPaged = mDataRequest.isPagingEnabled();

        for (DataRequest.Column columnRequest : mDataRequest.getColumns()) {
            final ColumnName columnName = columnRequest.getColumnName();

            if (!columnRequest.isPagingEnabled()) {
                completelyPaged = false;

                // Do not include max-versions from columns with paging enabled:
                largestMaxVersions = Math.max(largestMaxVersions, columnRequest.getMaxVersions());
            }

            if (columnName.isFullyQualified()) {
                addColumn(get, columnName);
            } else {
                get.addFamily(columnName.getFamilyBytes());

            }
            columnFilters.addFilter(toFilter(columnRequest));
        }

        if (completelyPaged) {
            // All requested columns have paging enabled.
            Preconditions.checkState(largestMaxVersions == 1);

            // We just need to know whether a row has data in at least one of the requested columns.
            // Stop at the first valid key using AND(columnFilters, FirstKeyOnlyFilter):
            get.setFilter(new FilterList(
                FilterList.Operator.MUST_PASS_ALL, columnFilters, new FirstKeyOnlyFilter()));
        } else {
            get.setFilter(columnFilters);
        }

        return get
            .setTimeRange(mDataRequest.getMinTimestamp(), mDataRequest.getMaxTimestamp())
            .setMaxVersions(largestMaxVersions);
    }

    /**
     * Configures a Scan with the options specified on HBaseScanOptions.
     * Whenever an option is not specified on <code>scanOptions</code>,
     * the hbase default will be used instead.
     *
     * @param scan        The Scan to configure.
     * @param scanOptions The options to configure this Scan with.
     */
    private void configureScan(final Scan scan, final HBaseScanOptions scanOptions) {
        if (null != scanOptions.getClientBufferSize()) {
            scan.setBatch(scanOptions.getClientBufferSize());
        }
        if (null != scanOptions.getCacheBlocks()) {
            scan.setCacheBlocks(scanOptions.getCacheBlocks());
        }
    }

}
